package com.flow;

import lombok.SneakyThrows;

import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;

/**
 * 发布者发布数据是将数据发布给队列  JVM内建的缓冲区
 * 订阅者就是找队列或者缓冲区拿数据
 * publisher 发布者
 * subscriber 订阅者
 * subscription  订阅关系
 *
 *  JVM底层对于整个订阅发布关系做好了 异步+缓存区处理
 *  效果就是发布者有数据,订阅者就会拿到数据
 */
public class FlowDemo02 {
    public static void main(String[] args) {
        //1.定义一个发布者
        SubmissionPublisher<String> publisher = new SubmissionPublisher();

        //2.定义一个中间操作,给每一个元素加上一个haha前缀
        MyProcessor myProcessor = new MyProcessor();


        //3.定义一个订阅者  对发布者的数据感兴趣
        Flow.Subscriber<String> subscriber = new Flow.Subscriber<String>() {
            private Flow.Subscription subscription;

            @Override //在订阅时   onXxxx   在XXXX事件发生时,执行这个回调方法
            public void onSubscribe(Flow.Subscription subscription) {
                System.out.println(Thread.currentThread() + "订阅开始了: " + subscription);
                this.subscription = subscription;
                //从上游请求一个数据
                subscription.request(1);
            }

            @Override //在下一个元素到达时,执行这个回调  接收新的数据
            public void onNext(String item) {
                System.out.println(Thread.currentThread() + "订阅者,接收到数据: " + item);
                //再要一个数据
                subscription.request(1);
//                if (item.equals("p-7")){
//                    subscription.cancel();  //取消订阅
//                }
            }

            @Override  //在错误发生时,订阅者接收到错误
            public void onError(Throwable throwable) {
                System.out.println(Thread.currentThread() + "订阅者,接收到错误的信号: " + throwable);
            }

            @Override  //在完成时
            public void onComplete() {
                System.out.println(Thread.currentThread() + "订阅者,接收完成信号");
            }
        };


        //4. 将发布者与处理器绑定在一起  再将处理器与订阅者绑定在一起
        //绑定操作: 发布者记住了所有订阅者都有谁,有数据后,把数据推给所有订阅者
        publisher.subscribe(myProcessor);
        myProcessor.subscribe(subscriber); //链表关系绑定出责任链

//        publisher.subscribe(subscriber);

        //发布者发布数据
        for (int i = 0; i < 10; i++) {
            System.out.println("发布数据的线程: " + Thread.currentThread());
            //发布者发布10条数据   publisher发布的数据在它的buffer区
            publisher.submit("p-" + i);
//            if (i >= 9){
//                publisher.closeExceptionally(new Exception("数据大于9,数据有误"));
//            }else {
//                publisher.submit("p-" + i);
//            }
        }

        //发布者通道关闭
        publisher.close();

        try {
            Thread.sleep(20000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
